package com.atguigu.chapter11;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.junit.Before;
import org.junit.Test;

/**
 * @ClassName: Flink11_Group_Window_Use
 * @Description:
 * @Author: kele
 * @Date: 2021/4/13 12:39
 *

 **/
public class Flink12_Over_Window_SQL {

    private StreamTableEnvironment tenv;
    private StreamExecutionEnvironment env;

    @Before
    public void before(){

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(1);

        tenv = StreamTableEnvironment.create(env);

        tenv.executeSql("create table sensor(" +
                "id string," +
                "ts bigint," +
                "vc int, " +
                "t as to_timestamp(from_unixtime(ts,'yyyy-MM-dd HH:mm:ss'))," +
                "watermark for t as t - interval '3' second)" +
                "with("
                + "'connector' = 'filesystem',"
                + "'path' = 'in/sensor.txt',"
                + "'format' = 'csv'"
                + ")");
    }

    @Test
    public void query(){

        tenv.sqlQuery("select \n" +
                "id,ts,vc,\n" +
                //rows between xxxx and xxx
                //注意 ： order by的时间必须是timestamp类型
/** 1 *///       "sum(vc) over(partition by id order by t rows between unbounded preceding and current row) vc_sum\n" +
/** 2 *///       "sum(vc) over(partition by id order by t rows between 1 preceding and current row) vc_sum\n" +
/** 3 */       "sum(vc) over(partition by id order by t range between unbounded preceding and current row) vc_sum\n" +
/** 4 *///       "sum(vc) over(partition by id order by t range between interval '1' second preceding and current row) vc_sum\n" +


                "from sensor")
                .execute()
                .print();

    }

    /**
     * 查询多个组合条件时，采用简写over的方式
     */
    @Test
    public void someQuery(){

        tenv.sqlQuery("select \n" +
                "id,ts,vc,\n" +
                "sum(vc) over w vc_sum\n" +
                "max(vc) over w vc_max\n" +
                "min(vc) over w vc_min\n" +
                "from sensor\n" +
                "window w as (partition by id order by t rows between unbounded preceding and current row)")
                .execute()
                .print();
    }


}
